package com.framework.rocketmq.producer;

import com.framework.rocketmq.config.RocketMQConfig;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

/**
 * 描述：
 * 生成者
 * @author xianggj
 * @Date 2021/10/13 18:20
 **/
@Slf4j
@Configuration
public class RocketMQProducer {

    @Autowired
    private RocketMQConfig rocketMQConfig;

    private DefaultMQProducer producer;

    @PostConstruct
    public void init() {
        producer = new DefaultMQProducer(rocketMQConfig.getProducerGroup());
        this.producer.setNamesrvAddr(rocketMQConfig.getNameSrvAddr());
        //如果需要同一个jvm中不同的this.producer往不同的mq集群发送消息，需要设置不同的instanceName
//      this.producer.setInstanceName(instanceName);
        if (rocketMQConfig.getMaxMessageSize() != null) {
            this.producer.setMaxMessageSize(rocketMQConfig.getMaxMessageSize());
        }
        if (rocketMQConfig.getSendMsgTimeout() != null) {
            this.producer.setSendMsgTimeout(rocketMQConfig.getSendMsgTimeout());
        }
        //如果发送消息失败，设置重试次数，默认为2次
        if (rocketMQConfig.getRetryTimesWhenSendFailed() != null) {
            this.producer.setRetryTimesWhenSendFailed(rocketMQConfig.getRetryTimesWhenSendFailed());
        }

        try {
            this.producer.start();
            log.info(String.format("this.producer is start ! groupName:[%s],namesrvAddr:[%s]"
                    , rocketMQConfig.getProducerGroup(), rocketMQConfig.getNameSrvAddr()));
        } catch (MQClientException e) {
            log.error(String.format("this.producer is error {} %s %s"
                    , e.getMessage(), e));
        }
        System.out.println("生产者启动完成");
    }

    /**
     * 异步发送消息
     * @param msg
     * @param sendCallback
     * @throws MQClientException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public void send(Message msg, SendCallback sendCallback) throws MQClientException, RemotingException, InterruptedException   {
        this.producer.send(msg, sendCallback);
    }

    /**
     * 发送定向消息
     * @param msg
     * @throws MQClientException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public void sendOneway(Message msg) throws MQClientException, RemotingException, InterruptedException   {
        this.producer.sendOneway(msg);
    }

    /**
     * 同步发送消息
     * @param msg
     * @throws MQClientException
     * @throws RemotingException
     * @throws InterruptedException
     */
    public SendResult send(Message msg) throws MQClientException, RemotingException, MQBrokerException, InterruptedException  {
        return this.producer.send(msg);
    }

    public void shutdown(){
        this.producer.shutdown();
    }
}
